package com.aotain.jupiter.flink;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.jupiter.util.FilterNullEnum;


public class FilterNullObject<T> implements FilterFunction<Tuple2<String,T>>{
	
	private FilterNullEnum filterenum = FilterNullEnum.ALLISNOTNULL;
	
	public FilterNullObject(FilterNullEnum filterEnum) {
		filterenum = filterEnum;
	}

	/**
	 * 
	 */
	private static final long serialVersionUID = -7771775930267194018L;

	@Override
	public boolean filter(Tuple2<String, T> arg0) throws Exception {
		// TODO Auto-generated method stub
		if(filterenum == FilterNullEnum.F0ISNOTNULL) {
			if(arg0.f0 == null) {
				return false;
			} else {
				return true;
			}
		} if (filterenum == FilterNullEnum.F1ISNOTNULL) {
			if(arg0.f1 == null) {
				return false;
			} else {
				return true;
			}
		} if (filterenum == FilterNullEnum.ALLISNOTNULL) {
			if(arg0.f0 == null || arg0.f1 == null) {
				return false;
			} else {
				return true;
			}
		}
		return false;
	}

}
